package com.ehotting.mqbd.core.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.ehotting.mqbd.core.helper.MQStringHelper;
import com.ehotting.mqbd.core.mq.consumer.action.ConsumerActionAble;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;


public class MessageConsumer {

    private static Logger log = LoggerFactory.getLogger(MessageConsumer.class);

    private MessageConsumerParam messageConsumerParam;

    private ConsumerActionAble consumerAction;

    private DefaultMQPushConsumer consumer;

    private String consumerGroup;

    public MessageConsumer(MessageConsumerParam messageConsumerParam,
                           String consumerGroup,
                           ConsumerActionAble consumerAction) {
        this.messageConsumerParam = messageConsumerParam;
        this.consumerGroup = consumerGroup;
        this.consumerAction = consumerAction;

    }

    public void start(){
        createDefaultMQPushConsumer();
        // 3. 订阅一个或多个 topic，并指定 tag 过滤条件，* 表示接收所有 tag 的消息
        try {
            String topicName = createTopic(consumerAction);
            log.info("topic:{}",topicName);
            String tag = createTag(consumerAction);
            log.info("tag:{}",tag);
            consumer.subscribe(topicName,tag);
            // 4. 注册回调接口来处理从Broker中收到的消息
            consumer.registerMessageListener(consumerAction.getMessageListener());
            // 5. 启动 PushTagAConsumer
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        log.info("The PushTagAConsumer Started.");
    }


    public void shutdown(){
        consumer.shutdown();
    }

    private void createDefaultMQPushConsumer(){
        log.info("-------------------------------------------------------------");
        log.info("Start init consumerGroup:{}",consumerGroup);
        // 1. 初始化 listener，并设置 listener group name
        consumer = new DefaultMQPushConsumer(consumerGroup);
        log.info("messageConsumerParam:{}", JSON.toJSONString(messageConsumerParam));
        // 2. 设置 NameServer 地址
        consumer.setNamesrvAddr(messageConsumerParam.getNameSrvAddr());
        // 消费者每次最多拉取多少条消息
        consumer.setPullBatchSize(messageConsumerParam.getPullBatchSize());
        // 每个线程一次性最多处理多少条消息
        consumer.setConsumeMessageBatchMaxSize(messageConsumerParam.getConsumeMessageBatchMaxSize());
        // 每条消息的最大重试次数
        consumer.setMaxReconsumeTimes(messageConsumerParam.getMaxReconsumeTimes());
        // 用于并发消费的最小/最大线程数
        consumer.setConsumeThreadMin(messageConsumerParam.getConsumeThreadMin());
        consumer.setConsumeThreadMax(messageConsumerParam.getConsumeThreadMax());

        consumer.setConsumeTimeout(messageConsumerParam.getConsumeTimeout());
        consumer.setMqClientApiTimeout(messageConsumerParam.getMqClientApiTimeout());

        /**
         * 先暂停消费一下，过SuspendCurrentQueueTimeMillis时间间隔后再重试一下，而不是放到重试队列里。
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.setSuspendCurrentQueueTimeMillis(10000);
        consumer.setMessageModel(MessageModel.CLUSTERING);
    }




    public static class MessageConsumerParam implements Cloneable, Serializable {

        // 2. 设置 NameServer 地址
        private String nameSrvAddr;
        // 消费者每次最多拉取多少条消息
        private Integer pullBatchSize;
        // 每个线程一次性最多处理多少条消息
        private Integer consumeMessageBatchMaxSize;
        // 每条消息的最大重试次数
        private Integer maxReconsumeTimes;
        // 用于并发消费的最小/最大线程数
        private Integer consumeThreadMin;
        private Integer consumeThreadMax;

        private Long consumeTimeout;

        private Integer mqClientApiTimeout;

        public String getNameSrvAddr() {
            return nameSrvAddr;
        }

        public void setNameSrvAddr(String nameSrvAddr) {
            this.nameSrvAddr = nameSrvAddr;
        }

        public Integer getPullBatchSize() {
            return pullBatchSize;
        }

        public void setPullBatchSize(Integer pullBatchSize) {
            this.pullBatchSize = pullBatchSize;
        }

        public Integer getConsumeMessageBatchMaxSize() {
            return consumeMessageBatchMaxSize;
        }

        public void setConsumeMessageBatchMaxSize(Integer consumeMessageBatchMaxSize) {
            this.consumeMessageBatchMaxSize = consumeMessageBatchMaxSize;
        }

        public Integer getMaxReconsumeTimes() {
            return maxReconsumeTimes;
        }

        public void setMaxReconsumeTimes(Integer maxReconsumeTimes) {
            this.maxReconsumeTimes = maxReconsumeTimes;
        }

        public Integer getConsumeThreadMin() {
            return consumeThreadMin;
        }

        public void setConsumeThreadMin(Integer consumeThreadMin) {
            this.consumeThreadMin = consumeThreadMin;
        }

        public Integer getConsumeThreadMax() {
            return consumeThreadMax;
        }

        public void setConsumeThreadMax(Integer consumeThreadMax) {
            this.consumeThreadMax = consumeThreadMax;
        }

        @Override
        public MessageConsumerParam clone() {
            return (MessageConsumerParam) SerializationUtils.clone(this);
        }

        public Long getConsumeTimeout() {
            return consumeTimeout;
        }

        public void setConsumeTimeout(Long consumeTimeout) {
            this.consumeTimeout = consumeTimeout;
        }

        public Integer getMqClientApiTimeout() {
            return mqClientApiTimeout;
        }

        public void setMqClientApiTimeout(Integer mqClientApiTimeout) {
            this.mqClientApiTimeout = mqClientApiTimeout;
        }
    }

    public String createTopic(ConsumerActionAble consumerAction){
        return MQStringHelper.createConsumerTopic(consumerAction);
    }

    public String createTag(ConsumerActionAble consumerAction){
        return MQStringHelper.createConsumerTag(consumerAction);
    }

}